package org.xarcher.emiya.utils

import java.util.Timer
import java.util.concurrent.Executors

import akka.actor.ActorRef
import akka.util.Timeout
import com.softwaremill.tagging.@@
import play.api.inject.ApplicationLifecycle

import scala.concurrent.{ ExecutionContext, Future, Promise }
import scala.util.{ Failure, Success, Try }
import scala.language.postfixOps

class FutureTimeLimited(
  val name: String,
  limitedActor: ActorRef,
  shutdownHook: ApplicationLifecycle)(implicit ec: ExecutionContext) {

  def limit[T](futureFunc: () => Future[T], key: String): Future[T] = {
    limit(futureFunc, 1, key)
  }

  def limit[T](
    futureFunc: () => Future[T],
    weight: Long,
    key: String): Future[T] = this.synchronized {
    val weight1 = weight
    val promise = Promise[T]

    val wrapper: FutureWrapper = new FutureWrapper {
      override lazy val runFuture: Future[Boolean] = {
        val endPromise = Promise[Boolean]
        Try { futureFunc() } match {
          case Success(f) =>
            f.andThen {
              case Success(r) =>
                promise.success(r)
                endPromise.success(true)
              case Failure(e) =>
                //e.printStackTrace
                promise.failure(e)
                endPromise.success(true)
            }
          case Failure(e) =>
            promise.failure(e)
            endPromise.success(true)
        }
        endPromise.future
      }

      override val weight = weight1
    }

    limitedActor ! TimeLimitedActor.AddWrapperModel(wrapper)

    val aa = promise.future

    {
      import java.util.TimerTask
      import akka.pattern.{ ask, pipe }
      import scala.concurrent.duration._
      val timer = new Timer()
      shutdownHook.addStopHook(() => Future.successful(Try { timer.cancel() }))
      timer.schedule(
        new TimerTask() {
          override def run(): Unit = {
            implicit val timeout = Timeout(5 seconds)

            if (!aa.isCompleted) {
              (limitedActor ? TimeLimitedActor.CountQueue)
                .mapTo[TimeLimitedActor.QueueSize]
                .map {
                  case TimeLimitedActor.QueueSize(size,
                    weightSum,
                    maxWeightSum) =>
                    println(
                      s"未完成的 actor 按时间限流逻辑-${name}-${key}，队列总长度：${size}，当前权重：${weightSum}，预设权重：${maxWeightSum}")
                }
            } else {
              //println(s"${name}-${key}-已完成")
              timer.cancel()
            }
          }
        },
        12000,
        12000)
    }

    aa.andThen {
      case Failure(e) =>
      //e.printStackTrace
    }
  }

}

class FutureTimeLimitedGen(
  limitedActor: () => ActorRef @@ TimeLimitedActor,
  shutdownHook: ApplicationLifecycle) {

  protected val ec =
    ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(8))
  shutdownHook.addStopHook(() => Future.successful(Try { ec.shutdown() }))

  def create(
    exceptWeight: Int,
    name: String,
    period: Int): FutureTimeLimited = {
    val actor = limitedActor()
    actor ! TimeLimitedActor.Start(exceptWeight, period)
    new FutureTimeLimited(name, actor, shutdownHook)(ec)
  }

}
